package com.ogawa.base.mqtt;

import android.annotation.SuppressLint;
import android.content.Context;
import android.text.TextUtils;


import com.blankj.utilcode.util.LogUtils;
import com.google.gson.Gson;
import com.ogawa.base.bean.MqttAccountBean;
import com.ogawa.base.constants.Constants;
import com.ogawa.base.bean.PublishBean;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.json.JSONObject;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

public class MqttManager implements MqttCallbackExtended {
    String TAG = "MqttManager";
    private MqttAndroidClient mqttClient;
    private MqttConnectOptions options;
    private ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);
    private String clientId;
    private String curSn;
    private MqttConnectListener mConnectListener;
    private List<MqttReceiveListener> mMqttReceiveListeners = new ArrayList<>();
    private List<MqttDeviceStateListener> mDeviceStateListeners = new ArrayList<>();
    private List<MqttSendMsgCallback> mMsgSendCallbacks = new ArrayList<>();
    private Gson mGson;
    private Context mContext;
    private String mqttAccount, mqttPsw;
    private List<MqttAccountBean> mqttAccountBeanList = new ArrayList<>();

    private MqttManager() {

    }

    public static MqttManager getInstance() {
        return InstanceHolder.INSTANCE;
    }


    public List<MqttAccountBean> getMqttAccountBeanList() {
        return mqttAccountBeanList;
    }

    public void addMqttAccountBean(MqttAccountBean mqttAccountBean){
        boolean isHave = false;
        for (int i = 0; i < mqttAccountBeanList.size(); i++) {
            MqttAccountBean tempBean = mqttAccountBeanList.get(i);
            if (tempBean.getSn().equals(mqttAccountBean.getSn())){
                isHave = true;
                tempBean.setPassword(mqttAccountBean.getPassword());
                tempBean.setUsername(mqttAccountBean.getUsername());
            }
        }
        if (!isHave){
            mqttAccountBeanList.add(mqttAccountBean);
        }
    }

    public String getCurSn() {
        return curSn;
    }

    public void setCurSn(String curSn) {
        this.curSn = curSn;
    }

    public void setMqttAccountBeanList(List<MqttAccountBean> mqttAccountBeanList) {
        this.mqttAccountBeanList = mqttAccountBeanList;
    }

    /**
     * 初始化MqttClient
     *
     * @param context
     */
    public void initClient(Context context) {
        clientId = "android" + (int) ((Math.random() * 9 + 1) * 10000);
        mContext = context.getApplicationContext();
        mqttClient = new MqttAndroidClient(mContext, Constants.MQTT_HOST, clientId);
        mqttClient.setCallback(this);
        options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(500);
        options.setConnectionTimeout(20);
        options.setKeepAliveInterval(20);
        options.setMaxReconnectDelay(5000);
        options.setMaxInflight(10);
        mGson = new Gson();
    }

    public String getClientId() {
        return clientId;
    }

    public void setMqttMsg(String mqttAccount, String mqttPsw) {
        this.mqttAccount = mqttAccount;
        this.mqttPsw = mqttPsw;
    }

    public String getMqttAccount() {
        return mqttAccount;
    }

    public String getMqttPsw() {
        return mqttPsw;
    }

    public void setMqttConnectListener(MqttConnectListener listener) {
        mConnectListener = listener;
    }

    public void clearMqttConnectListener() {
        mConnectListener = null;
    }



    public void registeronReceiveMqttMsg(MqttReceiveListener callback) {
        if (!mMqttReceiveListeners.contains(callback)) {
            mMqttReceiveListeners.add(callback);
        }
    }

    public void unRegisteronReceiveMqttMsg(MqttReceiveListener callback) {
        mMqttReceiveListeners.remove(callback);
    }

    public void registerMqttSendMsgCallback(MqttSendMsgCallback callback) {
        if (!mMsgSendCallbacks.contains(callback)) {
            mMsgSendCallbacks.add(callback);
        }
    }

    public void unRegisterMqttSendMsgCallback(MqttSendMsgCallback callback) {
        mMsgSendCallbacks.remove(callback);
    }

    public void registerMqttDeviceStateListener(MqttDeviceStateListener listener) {
        if (!mDeviceStateListeners.contains(listener)) {
            mDeviceStateListeners.add(listener);
        }
    }

    public void unRegisterMqttDeviceStateListener(MqttDeviceStateListener listener) {
        mDeviceStateListeners.remove(listener);
    }




    /**
     * 建立Mqtt连接
     */
    public boolean connect() {

        if (TextUtils.isEmpty(mqttAccount) || TextUtils.isEmpty(mqttPsw)) {
            return false;
        }
        LogUtils.e(TAG,"mqtt connect:" + mqttAccount + ":" + mqttPsw);
        options.setUserName(mqttAccount);
        options.setPassword(mqttPsw.toCharArray());
        pool.execute(() -> {
            try {
                mqttClient.connect(options);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
        return true;
    }

    /**
     * 订阅主题
     */
    public void subscribe(String[] topics, int[] qos, IMqttActionListener listener) {
        try {
            mqttClient.subscribe(topics, qos, mContext, listener);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    private void publish(String topic, int qos, String content) {
        if (!isConnect()) {
            return;
        }
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);
        pool.execute(() -> {
            try {
                mqttClient.publish(topic, message);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 发布消息,带附加值
     *
     * @param func
     * @param model
     */
    public void publish(String func, String model, String addition) {
        PublishBean bean = new PublishBean();
        bean.setFunctionCode(func);
        bean.setModelValue(model);
        bean.setClientId(clientId);
        bean.setAdditionalValue(addition);
        publish("cmd/down/" + bean.getSn(), 2, mGson.toJson(bean));
    }

    /**
     * 发布消息，不带附加值
     *
     * @param func
     * @param model
     */
    public void publish(String func, String model) {
        publish(func, model, null);
    }

    /**
     * 是否连接
     *
     * @return
     */
    public boolean isConnect() {
        try{
            return mqttClient.isConnected();
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 断开连接
     */
    public void disconnect() {
        if (isConnect()) {
            try {
                mqttClient.disconnect();
                onConnectChange(false);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    private void onConnectChange(boolean connect) {
        LogUtils.e("onConnectChange:" + connect);
        if (mConnectListener != null) {
            mConnectListener.onConnectStateChange(connect);
        }
    }

    /**
     * 连接丢失
     *
     * @param cause
     */
    @Override
    public void connectionLost(Throwable cause) {
        LogUtils.e("connectionLost");
        disconnect();
        onConnectChange(false);
    }

    /**
     * 收到信息
     *
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String msg = new String(message.getPayload());
//        LogUtils.e("接收消息：" + topic + "-----------messageArrived---------" + msg);
        if (mMqttReceiveListeners != null){
            for (int i = 0; i < mMqttReceiveListeners.size(); i++) {
                mMqttReceiveListeners.get(i).onReceiveMqttMsg(topic,message);
            }
        }

    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        try{
            byte[] payload = token.getMessage().getPayload();
            LogUtils.e("deliveryComplete：" + new String(payload));
        }catch (Exception e){
            e.printStackTrace();
            LogUtils.e("deliveryComplete：" + e.getMessage());
        }
    }

    /**
     * 建立连接成功
     *
     * @param reconnect
     * @param serverURI
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        LogUtils.e("connectComplete：" + reconnect);
        onConnectChange(true);
    }

    private static class InstanceHolder {
        @SuppressLint("StaticFieldLeak")
        private static final MqttManager INSTANCE = new MqttManager();
    }
}
